package mr

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"sort"
	"strconv"
	"time"
)
import "log"
import "net/rpc"
import "hash/fnv"

// for sorting by key.
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

// Map functions return a slice of KeyValue.
type KeyValue struct {
	Key   string
	Value string
}

// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	//time.Sleep(time.Second) //睡眠一秒再接任务
	for {
		args := TaskRequest{}
		reply := TaskResponse{}
		CallGetTask(&args, &reply)
		CurNMaptask := reply.CurNMapTask
		CurNReducetask := reply.CurNReducetask
		if reply.State == 0 && CurNMaptask >= 0 { //do map
			filename := reply.XTask.Filename
			id := strconv.Itoa(reply.XTask.Idmap)
			file, err := os.Open(filename)
			if err != nil {
				log.Fatalf("cannot open maptask %v", filename)
			}
			content, err := ioutil.ReadAll(file)
			if err != nil {
				log.Fatalf("cannot read %v", filename)
			}
			file.Close()
			kva := mapf(filename, string(content))
			num_reduce := reply.NReduceTask
			bucket := make([][]KeyValue, num_reduce)
			for _, kv := range kva {
				num := ihash(kv.Key) % num_reduce
				bucket[num] = append(bucket[num], kv)
			}
			for i := 0; i < num_reduce; i++ {
				tmp_file, error := ioutil.TempFile("", "mr-map-*")
				if error != nil {
					log.Fatalf("cannot open tmp_file")
				}
				enc := json.NewEncoder(tmp_file)
				err := enc.Encode(bucket[i])
				if err != nil {
					log.Fatalf("encode bucket error")
				}
				tmp_file.Close()
				out_file := "mr-" + id + "-" + strconv.Itoa(i)
				os.Rename(tmp_file.Name(), out_file)
			}
			CallTaskfin(&reply.XTask) //调用RPC的信道

			//fmt.Println("正在做map ", reply.XTask.Filename)
			//file, err := os.Open(reply.XTask.Filename)
			//if err != nil {
			//	log.Fatalf("cannot open maptask %v", reply.XTask.Filename)
			//}
			//content, err := ioutil.ReadAll(file)
			//if err != nil {
			//	log.Fatalf("cannot read %v", reply.XTask.Filename)
			//}
			//file.Close()
			//kva := mapf(reply.XTask.Filename, string(content)) //kv array。是kv数组，对切片words遍历，每一个word就会生成一个{word, “1”}键值对，然后传到kva里面，最后返回kva
			////写进mr-taskNumber-y文件中
			//WriteMapOutput(kva, reply.XTask.Idmap, reply.NReduceTask)
			////reply.MapTaskFin <- true
			//CallTaskfin() //调用RPC的信道
		} else if reply.State == 1 && CurNReducetask >= 0 { //do reduce
			num_map := reply.NMapTask
			id := strconv.Itoa(reply.XTask.Idreduce)
			intermediate := []KeyValue{}
			for i := 0; i < num_map; i++ {
				map_filename := "mr-" + strconv.Itoa(i) + "-" + id
				inputFile, err := os.OpenFile(map_filename, os.O_RDONLY, 0777)
				if err != nil {
					log.Fatalf("cannot open reducetask %v", map_filename)
				}
				dec := json.NewDecoder(inputFile)
				for {
					var kv []KeyValue
					if err := dec.Decode(&kv); err != nil {
						break
					}
					intermediate = append(intermediate, kv...)
				}
			}
			sort.Sort(ByKey(intermediate))
			out_file := "mr-out-" + id
			tmp_file, err := ioutil.TempFile("", "mr-reduce-*")
			if err != nil {
				log.Fatalf("cannot open tmp_file")
			}
			i := 0
			for i < len(intermediate) {
				j := i + 1
				for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
					j++
				}
				values := []string{}
				for k := i; k < j; k++ {
					values = append(values, intermediate[k].Value)
				}
				output := reducef(intermediate[i].Key, values)
				fmt.Fprintf(tmp_file, "%v %v\n", intermediate[i].Key, output)
				i = j
			}
			tmp_file.Close()
			os.Rename(tmp_file.Name(), out_file)
			CallTaskfin(&reply.XTask) //通过reply.Xtask可以知道当前完成了哪些id
			//fmt.Println("do reducetasking\n")
			//intermediate := []KeyValue{}
			//nmap := reply.NMapTask
			//for i := 0; i < nmap; i++ {
			//	mapOutFilename := "mr-" + strconv.Itoa(i) + "-" + strconv.Itoa(reply.XTask.Idreduce)
			//	//mr-i-id
			//	inputFile, err := os.OpenFile(mapOutFilename, os.O_RDONLY, 0777)
			//	if err != nil {
			//		log.Fatalf("cannot open reduceTask %v", mapOutFilename)
			//	}
			//	dec := json.NewDecoder(inputFile)
			//	for {
			//		var kv KeyValue
			//		if err := dec.Decode(&kv); err != nil {
			//			break
			//		}
			//		intermediate = append(intermediate, kv)
			//	}
			//}
			////排序写入reduceUutput文件,参考提供的mrsequential代码
			//sort.Sort(ByKey(intermediate))
			//oFilename := "mr-out-" + strconv.Itoa(reply.XTask.Idreduce)
			//tmpReduceOutFile, err := ioutil.TempFile("", "mr-reduce-*")
			////reduceOutputFile, err := os.OpenFile(oFilename, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0777)
			//if err != nil {
			//	log.Fatalf("cannot open file")
			//}
			////reduceOutputFile, _ := os.OpenFile("mr-out-0", os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0777)
			//i := 0
			//for i < len(intermediate) {
			//	j := i + 1
			//	for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			//		j++
			//	}
			//	values := []string{}
			//	//合并value进字符串数组
			//	for k := i; k < j; k++ {
			//		values = append(values, intermediate[k].Value)
			//	}
			//	output := reducef(intermediate[i].Key, values)
			//	print(output)
			//	// this is the correct format for each line of Reduce output.
			//	fmt.Fprintf(tmpReduceOutFile, "%v %v\n", intermediate[i].Key, output)
			//	i = j
			//}
			//tmpReduceOutFile.Close()
			//fmt.Printf(oFilename)
			//fmt.Printf("\n")
			//os.Rename(tmpReduceOutFile.Name(), oFilename)
			//CallTaskfin()
		} else if reply.State == 2 {
			break
		} else {
			continue
		}
		time.Sleep(time.Second)

	}

}

// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
func CallExample() {
	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.

	//rpc调用
	ok := call("Coordinator.Example", &args, &reply)
	if ok {
		// reply.Y should be 100.
		fmt.Printf("reply.Y %v\n", reply.Y)
	} else {
		fmt.Printf("call failed!\n")
	}
}
func CallGetTask(args *TaskRequest, reply *TaskResponse) {
	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.

	//rpc调用
	ok := call("Coordinator.GetTask", &args, &reply)
	if ok {
		fmt.Printf("reply.Name %v\n", reply.XTask.Filename)
	} else {
		fmt.Printf("call failed!\n")
	}
}
func CallTaskfin(args *Task) {
	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.

	// declare a reply structure.
	reply := ExampleReply{}
	//rpc调用
	ok := call("Coordinator.TaskFin", &args, &reply)
	if ok {
		fmt.Printf("call fin ok!")
	} else {
		fmt.Printf("call fin failed!\n")
	}
}

// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
